package server

import (
	"../../deps/lessgo/logger"
	"../../deps/lessgo/utils"
	"../../deps/lessgo/utilx"
	"../base"
	"../conf"
	"../net"
	"../state"
	"../store"
	"fmt"
	"math/rand"
	"strings"
	"time"
)

func LocalInitialize() error {

	//
	initBuckets := map[string]string{
		"sys": "System Directory",
		"log": "System Log Directory",
		"pub": "Public Directory",
	}
	for key, val := range initBuckets {

		var initBucket base.Bucket
		preRs, err := store.NodeGet("/sys/bs/" + key)
		if err == nil {
			_ = preRs.ReplyJson(&initBucket)
		}

		if initBucket.Name == "" {

			bucket := base.Bucket{
				Uuid:     utils.StringNewRand36(32),
				Name:     key,
				Desc:     val,
				DataSize: 1000,
				Created:  utilx.TimeNow("atom"),
				Updated:  utilx.TimeNow("atom"),
			}
			bucketjs, _ := utils.JsonEncode(bucket)

			ver, _ := store.Cmd("INCRBY", "sys:keeper:ver", 1).Int64()

			node := store.NodeProposal{
				Key:    "/sys/bs/" + key,
				Val:    bucketjs,
				VerSet: uint64(ver),
			}
			store.NodeSet(&node) // TODO
		}
	}

	return nil
}

func LocalTrackerStart() {

	rs, _ := store.Cmd("KEYS", "sys:peers:*").List()

	for _, v := range rs {

		pr, err := store.Cmd("HGETALL", v).Hash()
		if err != nil {
			continue
		}

		hostid := strings.ToLower(pr["hostid"])
		if len(hostid) < 8 {
			continue
		}
		url := fmt.Sprintf("%s://%s:%s", pr["proto"], pr["addr"], pr["port"])

		state.Locker.Lock()
		if _, ok := state.Peers[hostid]; !ok {

			state.Peers[hostid] = &state.Peer{
				Peer: base.Peer{
					Id:    hostid,
					Addr:  pr["addr"],
					Port:  pr["port"],
					Proto: pr["proto"],
				},
				Url: url,
			}

			state.PeerIds = append(state.PeerIds, hostid)
		}
		state.Locker.Unlock()
	}

	go func() {

		for {

			time.Sleep(2e9)

			leader := store.Cmd("GET", "sys:leader").String()

			// broadcast self-info
			psi := utilx.SysinfoGet()
			jpsi, _ := utils.JsonEncode(psi)

			msg := map[string]string{
				"action":  base.ActionPeerCast,
				"peerid":  conf.Config.HostId,
				"content": jpsi,
			}
			if leader != "" {
				net.PeerSend(leader, msg)
			} else if rand.Intn(8) == 0 {
				net.PeersSend(msg)
			}

			// Paxos::P1a
			// try to become a leader
			if len(leader) == 0 && len(state.Peers) > 0 {

				logger.Print("warn", "Peer."+conf.Config.HostId+" Leader Election Start")

				// Paxos::P2c
				if ledlock := store.Cmd("GET", "tmp:peer:newleader").String(); len(ledlock) == 0 {

					num := store.Cmd("INCRBY", "sys:keeper:ver", 1).String()

					store.Cmd("SETEX", "tmp:peer:newleader", 3, num)
					store.Cmd("DEL", "tmp:peer:newleader:value")

					msg := map[string]string{
						"action":          base.ActionLeaderNew,
						"ProposalVersion": num,
						"ProposalContent": conf.Config.HostId,
					}
					net.PeersSend(msg)
				}
			}

			// Leader Cast
			if leader != "" && leader == conf.Config.HostId {

				num := store.Cmd("GET", "sys:keeper:ver").String()

				started := store.Cmd("GET", "sys:keeper:started").String()
				if len(started) < 10 {
					started = utilx.TimeNow("atom")
				}
				store.Cmd("SETEX", "sys:keeper:started", 30, started)

				peers := []string{}
				for _, v := range state.Peers {
					peers = append(peers, fmt.Sprintf("1,%s,%s,%s,%s", v.Id, v.Addr, v.Port, v.Proto))
				}

				if len(peers) > 0 {
					msg := map[string]string{
						"action":  base.ActionLeaderCast,
						"leader":  conf.Config.HostId,
						"version": num,
						"started": started,
						"peers":   strings.Join(peers, ";"),
					}
					net.PeersSend(msg)
				}
			}
		}
	}()
}
